package com.luych.toolbox.data.service.job;

import com.luych.toolbox.common.enums.DataAgentStatus;
import com.luych.toolbox.common.enums.DataSourceType;
import com.luych.toolbox.common.event.base.service.publisher.WebSocketEventPublisher;
import com.luych.toolbox.common.event.base.service.publisher.WebSocketLetterPublisher;
import com.luych.toolbox.common.event.base.service.publisher.WebSocketNoticePublisher;
import com.luych.toolbox.common.feign.param.gateway.center.WebSocketEvent;
import com.luych.toolbox.common.service.tool.Cmd;
import com.luych.toolbox.data.service.constant.Constant;
import com.luych.toolbox.data.service.dao.DataAgentDao;
import com.luych.toolbox.data.service.entity.DataAgent;
import org.quartz.JobExecutionContext;
import org.quartz.JobExecutionException;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.cloud.bus.BusProperties;
import org.springframework.context.ApplicationEventPublisher;
import org.springframework.scheduling.quartz.QuartzJobBean;

public class DataAgentHealthCheckJob extends QuartzJobBean {
    private static final Logger LOGGER = LoggerFactory.getLogger(DataAgentHealthCheckJob.class);

    private static final String CMD_MYSQL_EXEC_TEST = "mysql --version > /dev/null";
    private static final String CMD_MYSQL_DUMP_TEST = "mysqldump --version > /dev/null";
    private static final String CMD_DM_EXEC_TEST = "disql help > /dev/null";
    private static final String CMD_DM_IMPT_TEST = "dimp help > /dev/null";
    private static final String CMD_DM_EXPT_TEST = "dexp help > /dev/null";

    private final BusProperties busProperties;
    private final ApplicationEventPublisher applicationEventPublisher;
    private final DataAgentDao dataAgentDao;

    public DataAgentHealthCheckJob(BusProperties busProperties,
                                   ApplicationEventPublisher applicationEventPublisher,
                                   DataAgentDao dataAgentDao) {
        this.busProperties = busProperties;
        this.applicationEventPublisher = applicationEventPublisher;
        this.dataAgentDao = dataAgentDao;
    }

    @Override
    protected void executeInternal(JobExecutionContext context) throws JobExecutionException {
        String id = context.getJobDetail().getJobDataMap().getString("id");

        WebSocketLetterPublisher letterPublisher = new WebSocketLetterPublisher(busProperties, applicationEventPublisher);
        WebSocketNoticePublisher noticePublisher = new WebSocketNoticePublisher(busProperties, applicationEventPublisher);
        WebSocketEventPublisher eventPublisher = new WebSocketEventPublisher(busProperties, applicationEventPublisher);

        DataAgent dataAgent = dataAgentDao.findById(id).orElseThrow(() ->
                new JobExecutionException(Constant.DATA_AGENT_NOT_FOUND));
        DataAgentStatus status;
        LOGGER.info("start to check data agent {} status...", dataAgent.getHost());
        try {
            if (DataSourceType.MYSQL.equals(dataAgent.getType())) {
                Cmd.executeRemote(dataAgent.getHost(), dataAgent.getPort(), dataAgent.getUser(), dataAgent.getPass(), CMD_MYSQL_EXEC_TEST);
                Cmd.executeRemote(dataAgent.getHost(), dataAgent.getPort(), dataAgent.getUser(), dataAgent.getPass(), CMD_MYSQL_DUMP_TEST);
            } else if (DataSourceType.DM.equals(dataAgent.getType())) {
                Cmd.executeRemote(dataAgent.getHost(), dataAgent.getPort(), dataAgent.getUser(), dataAgent.getPass(), CMD_DM_EXEC_TEST);
                Cmd.executeRemote(dataAgent.getHost(), dataAgent.getPort(), dataAgent.getUser(), dataAgent.getPass(), CMD_DM_IMPT_TEST);
                Cmd.executeRemote(dataAgent.getHost(), dataAgent.getPort(), dataAgent.getUser(), dataAgent.getPass(), CMD_DM_EXPT_TEST);
            }
            status = DataAgentStatus.NORMAL;
            LOGGER.info("data agent {} connect success", dataAgent.getHost());
        } catch (Exception e) {
            LOGGER.error("data agent {} connect error: {}", dataAgent.getHost(), e.getMessage());
            status = DataAgentStatus.ABNORMAL;
        }
        if (!dataAgent.getStatus().equals(status)) {
            String message = "数据代理连接状态通知";
            if (dataAgent.getStatus() != DataAgentStatus.ABNORMAL && status == DataAgentStatus.ABNORMAL) {
                String content = "数据代理[" + dataAgent.getName() + "]连接异常，请注意修复！";
                letterPublisher.error(dataAgent.getOwnerId(), message, content);
                noticePublisher.error(dataAgent.getOwnerId(), content);
                eventPublisher.send(new WebSocketEvent(WebSocketEvent.Type.DATA_AGENT_ABNORMAL, dataAgent.getId()));
            } else if (dataAgent.getStatus() != DataAgentStatus.NORMAL && status == DataAgentStatus.NORMAL) {
                String content = "数据代理[" + dataAgent.getName() + "]连接正常。";
                letterPublisher.success(dataAgent.getOwnerId(), message, content);
                noticePublisher.success(dataAgent.getOwnerId(), content);
                eventPublisher.send(new WebSocketEvent(WebSocketEvent.Type.DATA_AGENT_NORMAL, dataAgent.getId()));
            }

            dataAgent.setStatus(status);
            dataAgentDao.save(dataAgent);
        }

    }
}
